package com._51doit.spark07

import java.security.MessageDigest

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object UrlMd5demo {
  def md5Encoding(password: String): String = { // 得到一个信息摘要器
    val digest = MessageDigest.getInstance("md5")
    val result = digest.digest(password.getBytes)
    val buffer = new StringBuffer
    // 把没一个byte 做一个与运算 0xff;
    for (b <- result) { // 与运算
      val number = b & 0xff
      val str = Integer.toHexString(number)
      if (str.length == 1) buffer.append("0")
      buffer.append(str)
    }
    // 标准的md5加密后的结果
    buffer.toString
  }


  def main(args: Array[String]): Unit = {
    /*val md5test: String = md5Encoding("01hn.com")
    println(md5test)
    sys.exit(1)*/

    val conf: SparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getSimpleName)
    val sc: SparkContext = new SparkContext(conf)

    val urlrdd: RDD[String] = sc.textFile("D:\\07spark\\spark-day04\\作业题\\作业2\\40690.txt")
    val md5date: RDD[String] = sc.textFile("D:\\07spark\\spark-day04\\作业题\\作业2\\url.db1000")


    val res0: RDD[(String, String)] = urlrdd.map(um => {

      val arrs: Array[String] = um.split("\t#\t")
      val urldata: Array[String] = arrs(0).split("//")

      val urls: String = urldata(1)
      val types: String = arrs(1)

      val md5url: String = md5Encoding(urls).substring(0, 14)

      (md5url.trim, if (types.length == 1) "0" + types else types)

    })
   // res0.foreach(println)

    val res1: collection.Map[String, String] = res0.collectAsMap()
    val res2: Broadcast[collection.Map[String, String]] = sc.broadcast(res1)


    val res3: RDD[String] = md5date.map(mm => {
      val md5: String = mm.substring(0, 14)
      val typess: String = mm.substring(14)

      val newtypes: String = res2.value.getOrElse(md5, typess)


      md5 ++ newtypes

    })
    res3.foreach(println)



sc.stop()





  }




}
